package com.gizwits.kafkaProducer

import java.util.Properties

import kafka.producer._
object KafkaWordCountProducer {

  def main(args: Array[String]) {
    /*  if (args.length < 4) {
      System.err.println("Usage: KafkaWordCountProducer <metadataBrokerList> <topic> " +
        "<messagesPerSec> <wordsPerMessage>")
      System.exit(1)
    }*/

    // val Array(brokers, topic, messagesPerSec, wordsPerMessage) = args
    /**
     * (brokers, topic, messagesPerSec, wordsPerMessage
     * --broker-list s60:9092 --topic test
     */

    // Produces some random words between 1 and 10
    // Zookeper connection properties   10  100
    val brokers = "s60:9092"
    val topic = "test"
    val messagesPerSec = 10
    val wordsPerMessage = 10
    val props = new Properties()
    props.put("metadata.broker.list", brokers)
    props.put("serializer.class", "kafka.serializer.StringEncoder")

    val config = new ProducerConfig(props)
    val producer = new Producer[String, String](config)

    // Send some messages
    while (true) {
      val messages = (1 to messagesPerSec.toInt).map {
        messageNum =>
        val str = (1 to wordsPerMessage.toInt).map(x => scala.util.Random.nextInt(10).toString)
          .mkString(" ")

        new KeyedMessage[String, String](topic, str)
      }.toArray
      // send   messages
      producer.send(messages: _*)
      System.out.println(messages.toList)
      Thread.sleep(2000)
    }
  }


}

